package com.sunzm.spark.streaming

import org.apache.commons.lang3.time.DateFormatUtils
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

object TansformDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName(s"${this.getClass.getSimpleName.stripSuffix("$")}")
      .setMaster("local[*]")

    val ssc = new StreamingContext(conf, Durations.seconds(5))

    val lines = ssc.socketTextStream("82.156.210.70", 9999)

    val wordCounts: DStream[(String, Int)] = lines.transform(rdd => {

      rdd.flatMap(_.split(","))
        .map((_, 1))
        .reduceByKey(_ + _)

    })

    //wordCounts.print()

    wordCounts.foreachRDD((rdd, time) => {
      val milliseconds = time.milliseconds

      val timeStr = DateFormatUtils.format(milliseconds, "yyyy-MM-dd HH:mm:ss")

      rdd.foreach(line => {
        println(s"${timeStr}: ${line}")
      })

    })

    ssc.start()
    ssc.awaitTermination()
  }
}
